/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */

package lime2.ca.treeplain;

import lime2.ca.*;
import lime2.*;

import java.io.*;
import java.net.*;
import java.util.HashSet;
import java.util.List;
import java.util.logging.Logger;

/**
 * A TCP/UDP based communication adapter. It uses TCP for unicast communication
 * and UDP for multicast.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

public class TreePlainCA implements ICommunicationAdapter {
	// datagrams sender socket
	static DatagramSocket ssock;

	// multicast socket
	static MulticastSocket msock;

	// multicast group address
	static InetAddress maddr;

	// multicast group port
	static int mport;

	// internal queue for incoming remote ops
	static Queue queue;

	// unicast receiver
	static TreePlainUnicastReceiver pur;

	// multicast receiver
	static TreePlainMulticastReceiver pmr;
	
	// mesh info multicast receiver
	static MeshInfoReceiver mir;

	// detector
	static TreePlainDetector pd;
	
	// mesh information broadcaster
	static MeshInfoBroadcaster mib;

	/**
	 * receive socket for receive mesh information
	 */
	static MulticastSocket rmSock;

	@SuppressWarnings("unused")
	private Logger LOG = Logger.getLogger(this.getClass().getName());

	static List nextHops;

	/**
	 * This method must be called before the communication adapter is used; it
	 * performs all the necessary initializations (i.e. starting thread/s,
	 * establishing permanent connections, etc..)
	 * 
	 * @throws LimeCommunicationException
	 */
	public void init() throws LimeCommunicationException {
		// temporary variable
		int uport = LimeServer.getServer().getUnicastPort();

		// initializations
		mport = LimeServer.getServer().getMulticastPort();
		maddr = LimeServer.getServer().getMulticastID().getHost();
		queue = new Queue();

		// creates receiving unicast socket
		ServerSocket usock = null;
		try {
			usock = new ServerSocket(uport);
		} catch (IOException ioe) {

			throw new LimeCommunicationException(
					"can't listen on unicast port (TCP " + uport + ")", ioe,
					Failures.CRITICAL);
		}

		// creates datagrams sender socket
		ssock = null;
		try {
			ssock = new DatagramSocket(uport);
		} catch (IOException ioe) {
			System.out.println(ioe.getMessage());
			throw new LimeCommunicationException("can't use unicast port (UDP "
					+ uport + ")", ioe, Failures.CRITICAL);
		}

		// creates receiving multicast socket
		msock = null;
		try {
			msock = new MulticastSocket(mport);
		} catch (IOException ioe) {
			throw new LimeCommunicationException(
					"can't listen to multicast port " + maddr + "(" + mport + ")", ioe,
					Failures.CRITICAL);
		}
		
		// creates mesh information receiving multicast socket
		rmSock = null;
//		msock = null;
		try {
			rmSock = new MulticastSocket(mport + 1);
		} catch (IOException ioe) {
			throw new LimeCommunicationException(
					"can't listen to multicast port " + maddr + "(" + (mport + 1) + ")", ioe,
					Failures.CRITICAL);
		}

		// threading (.. please note that instances of Queue are internally
		// synchronized to prevent concurrent access)
		pur = new TreePlainUnicastReceiver(usock);
		pmr = new TreePlainMulticastReceiver(msock, ssock, uport);
		mir = new MeshInfoReceiver(rmSock);
		pur.start();
		pmr.start();
		mir.start();

		// creates detector
		pd = new TreePlainDetector();
		
		// creates information broadcaster
		mib = new MeshInfoBroadcaster();
	}

	/**
	 * Enables this host to join the system. The serializable object may serve
	 * to provide startup information for the joining process (i.e. a host to
	 * attach to in a dispatching tree). Please note that communication
	 * exceptions are not thrown; when this method is invoked, it is guaranteed
	 * that the LimeServer will join the system as soon as possible.
	 * 
	 * @param info
	 *            startup information for the joining process
	 */
	public void joinSystem(Serializable info) {
		pd.enable();
		mib.enable();
	}

	/**
	 * Enables this host to leave the system. Please note that communication
	 * exceptions are not thrown; when this method is invoked, it is guaranteed
	 * that the LimeServer will eventually leave the system.
	 */
	public void leaveSystem() {
		pd.disable();
		mib.disable();
	}

	/**
	 * Declares that from now on this node is interested in everything that
	 * concerns a specific tuplespace. Please note that communication exceptions
	 * are not thrown; when this method is invoked, it is guaranteed that the
	 * LimeServer will join the group as soon as possible.
	 * 
	 * @param name
	 *            name of the tuplespace we are interested in
	 */
	public void joinGroup(String name) {
		// does nothing
	}

	/**
	 * Declares that from now on this node is uninterested in everything that
	 * concerns a specific tuplespace. Please note that communication exceptions
	 * are not thrown; when this method is invoked, it is guaranteed that the
	 * LimeServer will eventually leave the group.
	 * 
	 * @param name
	 *            name of the tuplespace we are no more interested in
	 */
	public void leaveGroup(String name) {
		// does nothing
	}

	/**
	 * Sends an operation request to all the group members.
	 * 
	 * @param name
	 *            name of the destination tuplespace
	 * @param d
	 *            operation descriptor
	 * @throws LimeCommunicationException
	 */
	public void sendOp(String name, RemoteOp d)
			throws LimeCommunicationException {
		sendGroup(d);
	}

	/**
	 * Sends an operation request to a specific host.
	 * 
	 * @param host
	 *            destination host
	 * @param d
	 *            operation descriptor
	 * @throws LimeCommunicationException
	 */
	public void sendOp(LimeServerID host, RemoteOp d)
			throws LimeCommunicationException {
		send(host, d);
	}

	/**
	 * Sends a reaction registration request to all group members.
	 * 
	 * @param name
	 *            name of the destination tuplespace
	 * @param d
	 *            operation descriptor
	 * @throws LimeCommunicationException
	 */
	public void sendReaction(String name, RemoteOp d)
			throws LimeCommunicationException {
		sendGroup(d);
	}

	/**
	 * Sends a reaction registration request to a specific host.
	 * 
	 * @param host
	 *            destination host
	 * @param d
	 *            operation descriptor
	 * @throws LimeCommunicationException
	 */
	public void sendReaction(LimeServerID host, RemoteOp d)
			throws LimeCommunicationException {
		send(host, d);
	}

	/**
	 * Allows to bind a message to a previous operation request (i.e, to gather
	 * results).
	 * 
	 * @param host
	 *            destination host
	 * @param d
	 *            operation descriptor
	 * @throws LimeCommunicationException
	 */
	public void sendReply(LimeServerID host, RemoteOp d)
			throws LimeCommunicationException {
		send(host, d);
	}

	/**
	 * Returns a received RemoteOp (FIFO order, blocking).
	 * 
	 * @return RemoteOp
	 */
	public RemoteOp getNextOp() {
		return (RemoteOp) queue.remove();
	}

	/**
	 * Deactivates the communication layer.
	 */
	public void shutdown() {
		pur.shutdown();
		pmr.shutdown();
	}

	/**
	 * Returns whether an host is engaged.
	 * 
	 * @param host
	 * @return <code>true</code> or <code>false</code>
	 */
	public boolean queryHost(LimeServerID host) {
		return pd.queryHost(host);
	}

	/**
	 * Adds a new (discovered) host.
	 * 
	 * @param host
	 */
	public void addHost(LimeServerID host) {
		pd.addHost(host);
	}

	/**
	 * Removes a (probable) dead host.
	 * 
	 * @param host
	 */
	public void removeHost(LimeServerID host) {
		pd.removeHost(host);
	}

	/**
	 * Send an operation to a collection of targets through multicast tree
	 * @param op the operation to be sent
	 * @param hosts where the operation to be sent
	 * @return the hosts array to which this multicast directly sends
	 * @author Xing Jiankuan
	 */
	public HashSet<LimeServerID> multicast(LimeServerID[] hosts, RemoteOp op) throws LimeCommunicationException {
		if(hosts == null || hosts.length == 0) {
			return multicast(op);
		}
		
		Mesh mesh = Mesh.getInstance();
		
		//Fill the multicast info with mesh maintained order
		CastInfo opInfo = mesh.createCastInfo(hosts);
		assert(opInfo != null);
		op.setCastInfo(opInfo);
		
		//Send through tree
		return sendbyTree(op);
	}
	
	/**
	 * Send an operation to all engaged hosts through multicast tree
	 * @param op the operation to be sent
	 * @author Xing Jiankuan
	 */
	public HashSet<LimeServerID> multicast(RemoteOp op) throws LimeCommunicationException {
		Mesh mesh = Mesh.getInstance();
		
		CastInfo opInfo = mesh.createCastInfoWithAll();
		assert(opInfo != null);
		op.setCastInfo(opInfo);

		//Send through tree
		HashSet<LimeServerID> ret = sendbyTree(op);
		return ret;
	}
	
	// unicast send
	private void send(LimeServerID target, RemoteOp d)
			throws LimeCommunicationException {
		// initialization
		Connection c = null;

		// total failure ?
		try {
			c = Connection.get(target);
		} catch (IOException ioe) {
			throw new LimeCommunicationException(ioe, Failures.TOTAL);
		}

		// temporary variables
		boolean end = false;
		int retries = 0;

		// partial failure ?
		while (!end) {
			try {
				c.send(d);
				end = true;
			} catch (IOException ioe) {
				if (++retries >= LimeConstants.RETRIES)
					throw new LimeCommunicationException(ioe, Failures.PARTIAL);
			}
		}

		// irrelevant failure ?
		try {
			c.finished();
		} catch (IOException ioe) {
			throw new LimeCommunicationException(ioe, Failures.IRRELEVANT);
		}
	}

	// multicast send; we rely on internal synchronization of datatagram sockets
	private void sendGroup(RemoteOp d) throws LimeCommunicationException {
		// total failure ?
		if (ssock == null)
			throw new LimeCommunicationException(Failures.TOTAL);

		// temporary variables
		byte[] data = Serializer.serialize(d);
		boolean end = false;
		int retries = 0;

		// partial failure ? either the message can be delivered to all the
		// remote hosts, or it is delivered to none
		while (!end) {
			try {
				ssock.send(new DatagramPacket(data, data.length, maddr, mport));
				end = true;
			} catch (NullPointerException npe) {
				throw new LimeCommunicationException(npe, Failures.TOTAL);
			} // socket closed !
			catch (IOException ioe) {
				if (++retries >= LimeConstants.RETRIES)
					throw new LimeCommunicationException(ioe, Failures.PARTIAL);
			}
		}
	}

	/**
	 * multicast send, we rely on internal synchronization of datatagram sockets
	 * @param d
	 * @throws LimeCommunicationException
	 * @author Xing Jiankuan
	 */
	private HashSet<LimeServerID> sendbyTree(RemoteOp d) throws LimeCommunicationException {
		return d.getCastInfo().performMulticast(this, d);
	}
}